package transform;

import bean.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class TransDemo04_MultipleStreams {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputDS = env.readTextFile("D:\\CS\\MyDemo\\gmall-flink\\flink-demo\\src\\main\\resources\\sensor.txt");

        SingleOutputStreamOperator<SensorReading> dataStream = inputDS.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });


        // 1. 分流，按照温度值30度为界分为两条流
        // split (flink1.13 中被替换了... 用process)

//        SplitStream<SensorReading> splitStream = dataStream.split(new OutputSelector<SensorReading>() {
//            @Override
//            public Iterable<String> select(SensorReading value) {
//                return (value.getTemperature() > 30) ? Collections.singletonList("high") : Collections.singletonList("low");
//            }
//        });

        OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("side-output") {
        };

        SingleOutputStreamOperator<SensorReading> splitStream = dataStream.process(new ProcessFunction<SensorReading, SensorReading>() {
            @Override
            public void processElement(SensorReading sensorReading, Context context, Collector<SensorReading> collector) throws Exception {
                if (sensorReading.getTemperature() > 30) {
                    collector.collect(sensorReading);
                } else {
                    context.output(outputTag, sensorReading);
                }
            }
        });

        DataStream<SensorReading> sideOutput = splitStream.getSideOutput(outputTag);

//        splitStream.print("high");
//        sideOutput.print("low");


        // 2. 合流 connect，将高温流转换成二元组类型，与低温流连接合并之后，输出状态信息
        SingleOutputStreamOperator<Tuple2<String, Double>> warningStream = splitStream.map(new MapFunction<SensorReading, Tuple2<String, Double>>() {
            @Override
            public Tuple2<String, Double> map(SensorReading sensorReading) throws Exception {
                return new Tuple2<>(sensorReading.getId(), sensorReading.getTemperature());
            }
        });

        ConnectedStreams<Tuple2<String, Double>, SensorReading> connectedStreams = warningStream.connect(sideOutput);

        // CoMapFunction 合流map方法
        SingleOutputStreamOperator<Object> resultStream = connectedStreams.map(new CoMapFunction<Tuple2<String, Double>, SensorReading, Object>() {
            @Override
            public Object map1(Tuple2<String, Double> value) throws Exception {
                return new Tuple3<>(value.f0, value.f1, "high temp warning");
            }

            @Override
            public Object map2(SensorReading value) throws Exception {
                return new Tuple2<>(value.getId(), "normal");
            }
        });

//        resultStream.print();

        // 3. union联合多条流
        DataStream<SensorReading> unionStream = splitStream.union(sideOutput);
        unionStream.print();

        env.execute();

    }
}
